//
//package com.kalvan.core.mq.consumer;
//
//import lombok.extern.slf4j.Slf4j;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
//import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.stereotype.Component;
//import org.springframework.util.CollectionUtils;
//
//import javax.annotation.PostConstruct;
//import javax.annotation.PreDestroy;
//import java.util.LinkedList;
//import java.util.List;
//
///**
// * 用于创建消费者
// *
// * @author chenliang
// */
//@Slf4j
//@Component
//public class MqConsumer {
//
//    @Value("${rocketmq.namesrvAddr}")
//    private String namesrvAddr;
//
//    private List<DefaultMQPushConsumer> consumerContainer = new LinkedList<>();
//
//    @PostConstruct
//    public void start() {
//        //代收异步通知并发消费(控制线程池大小)
//        TransNotifyDsConsumer mqConsumer = new TransNotifyDsConsumer(50, 100, 20, 100);
//        consumerContainer.add(createConsumer("test-topic", mqConsumer));
//    }
//
//
//    /**
//     * 创建一个并发消息消费者
//     */
//    public DefaultMQPushConsumer createConsumer(String topic, MessageListenerConcurrently messageListener) {
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Group_" + topic);
//        log.info("MQ：启动消费者:{}，MQ地址：{}，消费主题：{}", this.getClass(), namesrvAddr, topic);
//        try {
//            consumer.setNamesrvAddr(namesrvAddr);
//            //实例名默认是ip+进程号
//            //consumer.setInstanceName("PushConsumer1");
//            // 从消息队列头开始消费
//            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//            // 集群消费模式
//            consumer.setMessageModel(MessageModel.CLUSTERING);
//            // 订阅主题
//            consumer.subscribe(topic, "*");
//            // 注册消息监听器
//            consumer.registerMessageListener(messageListener);
//            // 启动消费端
//            consumer.start();
//        } catch (MQClientException e) {
//            log.error("MQ：消费者{} 异常信息：{}-{}", this.getClass(), e.getResponseCode(), e.getErrorMessage());
//            throw new RuntimeException(e.getMessage(), e);
//        }
//        return consumer;
//    }
//
//
//    @PreDestroy
//    public void destroy() {
//        if (!CollectionUtils.isEmpty(consumerContainer)) {
//            for (DefaultMQPushConsumer consumer : consumerContainer) {
//                log.info("MQ：关闭消费者{}", consumer.getConsumerGroup());
//                consumer.shutdown();
//            }
//        }
//    }
//}
